package server

import (
	"bytes"
	"dag2.0/gat1400-proxy/cache"
	"dag2.0/gat1400-proxy/context"
	"dag2.0/gat1400-proxy/db"
	"dag2.0/gat1400-proxy/logger"
	"dag2.0/gat1400-proxy/model"
	"dag2.0/gat1400-proxy/model/constant"
	"dag2.0/gat1400-proxy/util"
	"dag2.0/gat1400-proxy/util/times"
	"encoding/json"
	"errors"
	"fmt"
	//json "github.com/json-iterator/go"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	URM_URL_EQUIPMENT            = "http://%s:/api/equipment/selectByParam"
	URM_URL_TOLLGATE             = "http://%s:/api/tollgate/getCompriseEqByParam"
	URM_URL_LANE                 = "http://%s:/api/lane/getByParam"
	RESOURCE_MANAGER_TASK_INFO   = "http://%s:%d:/api/dag/v2/resource/transmit/%s"
	RESOURCE_MANAGER_TASK_UPDATE = "http://%s:%d/api/dag/v2/resource/transmit/update"
	URM_PAGE_RESOURCE_BY_PARAM   = "http://%s:/api/resource/getPageResourceByParam"
	NOTIFICATION_EQ              = "notification_eq"
	NOTIFICATION_TG              = "notification_tg"
	NOTIFICATION_LANE            = "notification_lane"
)

func SubscribeNotification() {
	//启动订阅任务定时通知
	go func() {
		for {
			//订阅任务
			if cache.GetIns().SubscribeTasks.Len() > 0 {
				notificationAction()
			}
			time.Sleep(time.Second * time.Duration(30))
		}
	}()

	//启动资源变更通知
	//go watchResourceToNotify()
}
func notificationAction() {

	cache.GetIns().SubscribeTasks.Range(func(key string, value interface{}) (shouldContinue bool) {

		if _, ok := value.(*model.SubscribeLib); !ok {
			logger.LOG_ERROR(fmt.Sprintf("类型转换错误，需要*model.SubscribeLib 类型，类型为:%T", value))
			return true
		}
		subscribeLib := value.(*model.SubscribeLib)
		//只处理作为下级的订阅
		if _, ok := cache.GetIns().SubscribeManagers.Load(subscribeLib.ManagerID); !ok {
			return true
		}
		managerInterFace, _ := cache.GetIns().SubscribeManagers.Load(subscribeLib.ManagerID)
		manager := managerInterFace.(*model.SubscribeManagerEx)
		//视图库为上级（MANAGER_UP），表示此视图库的任务为上级订阅下级
		if manager.ManagerType != constant.MANAGER_UP {
			return true
		}
		//一个订阅id对应一个订阅任务
		resourceFlag, dataFlage := false, false
		//已处理完的订阅
		if subscribeLib.SubscribeStatus == 99 {
			return true
		}
		//0订阅中，1已取消订阅，2订阅到期，9未订阅,99已完成通知
		if subscribeLib.SubscribeStatus == 1 || subscribeLib.SubscribeStatus == 2 || subscribeLib.SubscribeStatus == 9 {
			return true
		}
		//判断订阅任务
		details := strings.Split(subscribeLib.SubscribeDetail, ",")
		for _, detail := range details {
			switch detail {
			//资源
			case constant.EQUIPMENT_LIST, constant.EQUIPMENT_STATUS, constant.TOLLGATE_LIST,
				constant.SINGLE_TOLLGATE_INFO, constant.LAND_LIST, constant.SINGLE_LAND_INFO, constant.HIK_ILLEGAL:
				resourceFlag = true
			//数据
			case constant.PERSON_INFO, constant.FACE_INFO, constant.VEHICLE_INFO, constant.NOMOTOR_INFO:
				dataFlage = true
			default:
				logger.LOG_WARN("不支持的订阅类型", detail)
			}
		}
		if resourceFlag {
			go NotificationResource(subscribeLib, nil, true)
		}
		if dataFlage {
			NotificationData(subscribeLib)
		}
		return true
	})
}

type DiffApe struct {
	UpdateEqs []*model.APE
	AddEqs    []*model.APE
	RemoveEqs []*model.APE
}
type DiffTollgate struct {
	UpdateTgs []*model.Tollgate
	AddTgs    []*model.Tollgate
	RemoveTgs []*model.Tollgate
}
type DiffLane struct {
	UpdateLanes []*model.Lane
	AddLanes    []*model.Lane
	RemoveLanes []*model.Lane
}
type DiffRes struct {
	UpdateEqs   []*model.APE
	UpdateTgs   []*model.Tollgate
	UpdateLanes []*model.Lane

	AddEqs   []*model.APE
	AddTgs   []*model.Tollgate
	AddLanes []*model.Lane

	RemoveEqs   []*model.APE
	RemoveTgs   []*model.Tollgate
	RemoveLanes []*model.Lane
}

func incrementLane(fullLanes []*model.Lane, lnPushes map[string]*model.DevicePush) *DiffLane {
	var fullLaneMap = make(map[string]*model.Lane, 0)
	var updateLanes = make([]*model.Lane, 0)
	var addLanes = make([]*model.Lane, 0)
	var removeLanes = make([]string, 0)
	for _, lane := range fullLanes {
		fullLaneMap[lane.UrmId] = lane
	}
	//判断增量
	for id, lnPush := range lnPushes {
		lane, ok := fullLaneMap[id]
		if !ok {
			//删除
			removeLanes = append(removeLanes, id)
		} else if lnPush.PushTime != lane.UrmCrateTime {
			//更新
			updateLanes = append(updateLanes, lane)
		}
	}
	//新增
	for id, urmLane := range fullLaneMap {
		_, ok := lnPushes[id]
		if !ok {
			addLanes = append(addLanes, urmLane)
		}
	}
	if len(removeLanes) == 0 {
		return &DiffLane{
			UpdateLanes: updateLanes,
			AddLanes:    addLanes,
			RemoveLanes: make([]*model.Lane, 0),
		}
	}
	//查询urm
	lanes, err := Deallane(removeLanes)
	if err != nil {
		logger.LOG_ERROR("查询urm车道失败 ，", err)
	}
	return &DiffLane{
		UpdateLanes: updateLanes,
		AddLanes:    addLanes,
		RemoveLanes: model.BuildLane(lanes),
	}
}

func incrementTollgate(fullTgs []*model.Tollgate, tgPushes map[string]*model.DevicePush) *DiffTollgate {
	var fullTgMap = make(map[string]*model.Tollgate, 0)
	var updateTgs = make([]*model.Tollgate, 0)
	var addTgs = make([]*model.Tollgate, 0)
	//var removeTgs = make([]*model.Tollgate, 0)
	var removeTgs = make([]string, 0)
	for _, tg := range fullTgs {
		fullTgMap[tg.UrmId] = tg
	}
	//判断增量
	for id, tgPush := range tgPushes {
		tg, ok := fullTgMap[id]
		if !ok {
			//删除
			removeTgs = append(removeTgs, id)
		} else if tgPush.PushTime != tg.UrmCrateTime {
			//更新
			updateTgs = append(updateTgs, tg)
		}
	}
	//新增
	for id, urmTg := range fullTgMap {
		_, ok := tgPushes[id]
		if !ok {
			addTgs = append(addTgs, urmTg)
		}
	}
	if len(removeTgs) == 0 {
		return &DiffTollgate{
			UpdateTgs: updateTgs,
			AddTgs:    addTgs,
			RemoveTgs: make([]*model.Tollgate, 0),
		}
	}
	//查询urm已删除的卡口
	urmTollgates, getTollResourceErr := Dealtollgate(removeTgs)
	if getTollResourceErr != nil {
		logger.LOG_ERROR("查询已删除的卡口失败 ", getTollResourceErr)
	}
	return &DiffTollgate{
		UpdateTgs: updateTgs,
		AddTgs:    addTgs,
		RemoveTgs: model.BuildTollgate(urmTollgates),
	}
}

func incrementApe(fullEqs []*model.APE, eqPushes map[string]*model.DevicePush) *DiffApe {
	var fullEqMap = make(map[string]*model.APE, 0)
	for _, ape := range fullEqs {
		fullEqMap[ape.UrmId] = ape
	}
	var updateEqs = make([]*model.APE, 0)
	var addEqs = make([]*model.APE, 0)
	var removeIds = make([]string, 0)
	//判断增量
	for id, aeq := range eqPushes {
		teq, ok := fullEqMap[id]
		//删除
		if !ok {
			//已删除的采集设备id集合
			removeIds = append(removeIds, id)
			// 卡口下的采集设备urm未返回更新时间 aeq.PushTime == 0
		} else if aeq.PushTime == 0 || aeq.PushTime != teq.UrmCrateTime {
			//更新
			updateEqs = append(updateEqs, teq)
		}
	}
	//新增
	for id, urmEquipment := range fullEqMap {
		_, ok := eqPushes[id]
		if !ok {
			addEqs = append(addEqs, urmEquipment)
		}
	}
	if len(removeIds) == 0 {
		return &DiffApe{
			UpdateEqs: updateEqs,
			AddEqs:    addEqs,
			RemoveEqs: make([]*model.APE, 0),
		}
	}
	//查询urm已删除的设备
	//todo 由于卡口下的采集设备urm未含有更新时间，全量推送，这里不需要判断卡口下的采集设备
	urmEquipments, getResourceErr := DealEquipment(removeIds)
	if getResourceErr != nil {
		logger.LOG_ERROR("查询urm失败")
	}
	return &DiffApe{
		UpdateEqs: updateEqs,
		AddEqs:    addEqs,
		RemoveEqs: model.BuildAPEFromUrmEquipments(urmEquipments),
	}
}
func fullApe(urmEquipments []*model.URMEquipment, urmTollgates []*model.URMTollgate) []*model.APE {
	var notificationAPEs = make([]*model.APE, 0)
	if urmEquipments != nil && len(urmEquipments) > 0 {
		//构造采集设备
		logger.LOG_INFO("绑定采集设备数量为: ", len(urmEquipments))
		notificationAPEs = append(notificationAPEs, model.BuildAPEFromUrmEquipments(urmEquipments)...)
		logger.LOG_INFO("构造通知采集设备数量为: ", len(notificationAPEs))
	}
	if urmTollgates != nil && len(urmTollgates) > 0 {
		//根据配置构造卡口下采集设备
		if context.GetBool("notification.dealTollgateEq") && urmTollgates != nil && len(urmTollgates) > 0 {
			logger.LOG_INFO("绑定卡口数量为: ", len(urmTollgates))
			notificationAPEs = append(notificationAPEs, model.BuildAPEFromUrmTollgate(urmTollgates)...)
			logger.LOG_INFO("构造通知卡口下采集设备数量为: ", len(notificationAPEs))
		}
	}
	return notificationAPEs
}
func fullTollgate(urmTollgates []*model.URMTollgate) []*model.Tollgate {
	var notificationTollgates = make([]*model.Tollgate, 0)
	if urmTollgates == nil || len(urmTollgates) == 0 {
		return notificationTollgates
	}
	logger.LOG_INFO("绑定卡口数量为: ", len(urmTollgates))
	notificationTollgates = append(notificationTollgates, model.BuildTollgate(urmTollgates)...)
	logger.LOG_INFO("构造通知卡口数量为: ", len(notificationTollgates))
	return notificationTollgates
}
func fullLane(urmTollgates []*model.URMTollgate) []*model.Lane {
	var notificationLanes = make([]*model.Lane, 0)
	if urmTollgates == nil || len(urmTollgates) == 0 {
		return notificationLanes
	}

	logger.LOG_INFO("绑定卡口数量为: ", len(urmTollgates))
	notificationLanes = append(notificationLanes, model.BuildLaneByTg(urmTollgates)...)
	logger.LOG_INFO("构造通知卡口下车道数量为: ", len(notificationLanes))
	return notificationLanes
}

// 处理资源的通知
func NotificationResource(n *model.SubscribeLib, subscribeManager *model.SubscribeManagerEx, increment bool) (*model.Response, error) {
	if n.OperateType == 1 {
		return nil, errors.New("该订阅已取消")
	}
	//获取视图库信息
	if _, ok := cache.GetIns().SubscribeManagers.Load(n.ManagerID); !ok {
		logger.LOG_ERROR("通知任务所属视图库id未备案")
		return nil, errors.New("通知任务所属视图库id未备案")
	}
	if subscribeManager == nil {
		subscribeManagerInterFace, _ := cache.GetIns().SubscribeManagers.Load(n.ManagerID)
		subscribeManager = subscribeManagerInterFace.(*model.SubscribeManagerEx)
		subscribeManager.ExecuteOperation = EXECUTE_OPERATION_ADD
	}
	details := strings.Split(n.SubscribeDetail, ",")
	notificationTypeMap := make(map[string]bool, 0)
	for _, detail := range details {
		switch detail {
		//采集设备
		case constant.EQUIPMENT_LIST, constant.EQUIPMENT_STATUS:
			notificationTypeMap[NOTIFICATION_EQ] = true
			//卡口
		case constant.TOLLGATE_LIST, constant.SINGLE_TOLLGATE_INFO, constant.HIK_ILLEGAL:
			notificationTypeMap[NOTIFICATION_TG] = true
			//车道
		case constant.LAND_LIST, constant.SINGLE_LAND_INFO:
			notificationTypeMap[NOTIFICATION_LANE] = true
		}
	}
	if len(subscribeManager.EquipmentIds) <= 0 && len(subscribeManager.TollgateIds) <= 0 {
		logger.LOG_WARN("此视图库无资源绑定，视图库为: ", subscribeManager.ManagerName)
		return nil, errors.New(fmt.Sprintf("此视图库无资源绑定,视图库为: %s", subscribeManager.ManagerName))
	}
	urmEquipments := make([]*model.URMEquipment, 0)
	urmTollgates := make([]*model.URMTollgate, 0)
	var getResourceErr error
	//卡口或车道或卡口下的采集设备
	if notificationTypeMap[NOTIFICATION_TG] || notificationTypeMap[NOTIFICATION_LANE] ||
		(notificationTypeMap[NOTIFICATION_EQ] && context.GetBool("notification.dealTollgateEq")) {
		urmTollgates, getResourceErr = Dealtollgate(subscribeManager.TollgateIds)
	}
	//采集设备
	if notificationTypeMap[NOTIFICATION_EQ] {
		//获取urm采集设备
		urmEquipments, getResourceErr = DealEquipment(subscribeManager.EquipmentIds)
		//Equipment 必填项校验，若为空，赋默认值
		checkEquipmentsAndCompletion(urmEquipments)
	}
	if getResourceErr != nil {
		logger.LOG_ERROR("获取资源信息失败", getResourceErr)
		return nil, getResourceErr
	}
	//构造通知资源
	diffApe, diffTollgate, diffLane := notificationResource(notificationTypeMap, urmEquipments, urmTollgates, increment, subscribeManager.ManagerID)
	if notificationTypeMap[NOTIFICATION_TG] {
		logger.LOG_WARN("ResourceNotify tollgate [add = ", len(diffTollgate.AddTgs), ", update = ", len(diffTollgate.UpdateTgs), ", delete = ", len(diffTollgate.RemoveTgs), "]")
	}
	if notificationTypeMap[NOTIFICATION_EQ] {
		logger.LOG_WARN("ResourceNotify ape [add = ", len(diffApe.AddEqs), ", update = ", len(diffApe.UpdateEqs), ", delete = ", len(diffApe.RemoveEqs), "]")
	}
	if notificationTypeMap[NOTIFICATION_LANE] {
		logger.LOG_WARN("ResourceNotify lane [add = ", len(diffLane.AddLanes), ", update = ", len(diffLane.UpdateLanes), ", delete = ", len(diffLane.RemoveLanes), "]")
	}
	subscribeNotificationModels := make([]*model.SubscribeNotificationModel, 0)

	//新增
	if len(diffApe.AddEqs)+len(diffTollgate.AddTgs)+len(diffLane.AddLanes) > 0 {
		subscribeNotificationModels = append(subscribeNotificationModels, splitNotificationResource(diffApe.AddEqs, diffTollgate.AddTgs, diffLane.AddLanes, n, EXECUTE_OPERATION_ADD)...)
	}
	//修改
	if len(diffApe.UpdateEqs)+len(diffTollgate.UpdateTgs)+len(diffLane.UpdateLanes) > 0 {
		subscribeNotificationModels = append(subscribeNotificationModels, splitNotificationResource(diffApe.UpdateEqs, diffTollgate.UpdateTgs, diffLane.UpdateLanes, n, EXECUTE_OPERATION_UPD)...)
	}
	//删除
	if len(diffApe.RemoveEqs)+len(diffTollgate.RemoveTgs)+len(diffLane.RemoveLanes) > 0 {
		//
		subscribeNotificationModels = append(subscribeNotificationModels, splitNotificationResource(diffApe.RemoveEqs, diffTollgate.RemoveTgs, diffLane.RemoveLanes, n, EXECUTE_OPERATION_DEL)...)
	}
	if nil == subscribeNotificationModels || len(subscribeNotificationModels) == 0 {
		_ = FlushSubscribeStatusCache(n, "99", "")
		logger.LOG_WARN("未构造出通知结构体")
		return nil, nil
	}
	//发送通知请求
	response, e := sendNotification(subscribeNotificationModels, n.ReceiveAddr, subscribeManager.SelfManagerID)
	if e != nil {
		return response, e
	}
	//设备发送成功后更改绑定资源列表成功时间
	e = UpdatePushRecords(diffApe, diffTollgate, diffLane, subscribeManager.ManagerID)
	if e != nil {
		return nil, e
	}
	//成功时改变内存与数据库的数据状态
	logger.LOG_WARN("发送设备通知成功，通知个数为：", len(subscribeNotificationModels))
	_ = FlushSubscribeStatusCache(n, "99", "")
	return nil, nil
}
func notificationResource(notificationTypeMap map[string]bool, eq []*model.URMEquipment, tg []*model.URMTollgate, increment bool, managerId string) (*DiffApe, *DiffTollgate, *DiffLane) {
	diffApe := &DiffApe{}
	diffTollgate := &DiffTollgate{}
	diffLane := &DiffLane{}
	for key := range notificationTypeMap {
		switch key {
		case NOTIFICATION_EQ:
			apes := fullApe(eq, tg)
			if !increment {
				diffApe.AddEqs = apes
				continue
			}
			//增量推送
			//查询已登记的成功推送记录
			records, err := db.SelectPushRecordByManagerAndType(managerId, model.RESOURCE_TYPE_EQUIPMENT)
			if err != nil {
				logger.LOG_ERROR("查询此视图库已推送成功的采集设备记录失败", err)
			}
			diffApe = incrementApe(apes, records)
		case NOTIFICATION_TG:
			tollgate := fullTollgate(tg)
			if !increment {
				diffTollgate.AddTgs = tollgate
				continue
			}
			//增量推送
			//查询已登记的成功推送记录
			records, err := db.SelectPushRecordByManagerAndType(managerId, model.RESOURCE_TYPE_TOLLGATE)
			if err != nil {
				logger.LOG_ERROR("查询此视图库已推送成功的卡口记录失败", err)
			}
			diffTollgate = incrementTollgate(tollgate, records)
		case NOTIFICATION_LANE:
			lane := fullLane(tg)
			if !increment {
				diffLane.AddLanes = lane
				continue
			}
			//增量推送
			//查询已登记的成功推送记录
			records, err := db.SelectPushRecordByManagerAndType(managerId, model.RESOURCE_TYPE_LANE)
			if err != nil {
				logger.LOG_ERROR("查询此视图库已推送成功的车道记录失败", err)
			}
			diffLane = incrementLane(lane, records)
		}
	}
	return diffApe, diffTollgate, diffLane
}

func sendNotification(subscribeNotificationModels []*model.SubscribeNotificationModel, addr, selfManagerID string) (*model.Response, error) {
	for _, subscribeNotificationModel := range subscribeNotificationModels {
		i := 1
		jsonData, err := json.Marshal(subscribeNotificationModel)
		if err != nil {
			logger.LOG_ERROR(err)
			continue
		}
		logger.LOG_WARN("订阅资源通知 url=", addr, " selfManagerId=", selfManagerID)
		logger.LOG_INFO("发送第", strconv.Itoa(i), "次通知, 通知结构为 ：", string(jsonData))
		res, _, _, e := util.Request(addr, http.MethodPost, map[string]string{
			"User-Identify": selfManagerID,
			"Content-Type":  "application/VIID+JSON",
		}, subscribeNotificationModel, nil)
		if e != nil {
			logger.LOG_ERROR("发送设备通知失败", e)
			return nil, e
		}
		response := model.AnalyzeRsponseContent(res)
		resJson, _ := json.Marshal(response)
		//失败
		if response == nil || response.ResponseStatusListObject == nil || response.ResponseStatusListObject.ResponseStatusObject == nil ||
			len(response.ResponseStatusListObject.ResponseStatusObject) <= 0 {
			logger.LOG_ERROR("发送设备通知失败 ", util.BytesString(resJson)) //等待下次重新发送
			return nil, errors.New("解析返回结构体失败")
		}
		for _, res := range response.ResponseStatusListObject.ResponseStatusObject {
			if res.StatusCode != model.OK {
				logger.LOG_ERROR("发送设备通知失败 ", util.BytesString(resJson)) //等待下次重新发送
				return response, errors.New("上级返回失败")
			}
		}
		i++
		logger.LOG_DEBUG("发送设备通知成功，返回数据为：", util.BytesString(resJson))
	}
	return nil, nil
}

func UpdatePushRecords(diffApes *DiffApe, diffTg *DiffTollgate, diffLane *DiffLane, managerId string) error {
	var updatePushRecords = make([]*model.DevicePush, 0)
	var deletePushRecords = make([]*model.DevicePush, 0)
	if len(diffApes.AddEqs) > 0 {
		for _, ape := range diffApes.AddEqs {
			updatePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   ape.UrmId,
				PushTime:     ape.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_EQUIPMENT,
				MangerId:     managerId,
			})
		}
	}
	if len(diffApes.UpdateEqs) > 0 {
		for _, ape := range diffApes.UpdateEqs {
			updatePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   ape.UrmId,
				PushTime:     ape.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_EQUIPMENT,
				MangerId:     managerId,
			})
		}
	}
	if len(diffApes.RemoveEqs) > 0 {
		for _, ape := range diffApes.RemoveEqs {
			deletePushRecords = append(deletePushRecords, &model.DevicePush{
				ResourceId:   ape.UrmId,
				PushTime:     ape.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_EQUIPMENT,
				MangerId:     managerId,
			})
		}
	}
	if len(diffTg.AddTgs) > 0 {
		for _, tg := range diffTg.AddTgs {
			updatePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   tg.UrmId,
				PushTime:     tg.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_TOLLGATE,
				MangerId:     managerId,
			})
		}
	}
	if len(diffTg.UpdateTgs) > 0 {
		for _, tg := range diffTg.UpdateTgs {
			updatePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   tg.UrmId,
				PushTime:     tg.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_TOLLGATE,
				MangerId:     managerId,
			})
		}
	}
	if len(diffTg.RemoveTgs) > 0 {
		for _, tg := range diffTg.RemoveTgs {
			deletePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   tg.UrmId,
				PushTime:     tg.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_TOLLGATE,
				MangerId:     managerId,
			})
		}
	}
	if len(diffLane.AddLanes) > 0 {
		for _, lane := range diffLane.AddLanes {
			updatePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   lane.UrmId,
				PushTime:     lane.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_LANE,
				MangerId:     managerId,
			})
		}
	}
	if len(diffLane.UpdateLanes) > 0 {
		for _, lane := range diffLane.UpdateLanes {
			updatePushRecords = append(updatePushRecords, &model.DevicePush{
				ResourceId:   lane.UrmId,
				PushTime:     lane.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_LANE,
				MangerId:     managerId,
			})
		}
	}
	if len(diffLane.RemoveLanes) > 0 {
		for _, lane := range diffLane.RemoveLanes {
			deletePushRecords = append(deletePushRecords, &model.DevicePush{
				ResourceId:   lane.UrmId,
				PushTime:     lane.UrmCrateTime,
				ResourceType: model.RESOURCE_TYPE_LANE,
				MangerId:     managerId,
			})
		}
	}
	//更新
	e := db.UpdateDevicePushTime(updatePushRecords, deletePushRecords)
	if e != nil {
		logger.LOG_ERROR("更新成功推送记录失败")
		return e
	}
	return nil
}
func checkEquipmentsAndCompletion(urmEquipments []*model.URMEquipment) {
	if len(urmEquipments) == 0 {
		logger.LOG_WARN("urm返回的设备为空")
		return
	}
	for _, eq := range urmEquipments {
		if len(eq.Model) == 0 {
			eq.Model = "0"
		}
		if len(eq.PlaceCode) == 0 {
			eq.PlaceCode = "0"
		}
	}
}

// 拆分通知请求
func splitNotificationResource(apes []*model.APE, tollgates []*model.Tollgate, lanes []*model.Lane,
	n *model.SubscribeLib, executeOperation int) []*model.SubscribeNotificationModel {
	var subscribeNotificationList = make([]*model.SubscribeNotification, 0)
	var subscribeNotificationModels = make([]*model.SubscribeNotificationModel, 0)
	if len(apes)+len(tollgates)+len(lanes) == 0 {
		logger.LOG_ERROR("未构造出通知结构体")
		return nil
	}
	batchSize := context.GetIntDefault("notification.batchSize", 2000)
	subscribeNotification := &model.SubscribeNotification{
		NotificationID:   util.GenerateSubscribeNotificationID(),
		SubscribeID:      n.SubscribeID,
		Title:            n.Title,
		TriggerTime:      times.Time2StrWithFormat(times.DateFormatNoSpan),
		InfoIDs:          n.ManagerID,
		ExecuteOperation: executeOperation,
	}
	//对于设备集合小于batchSize的通知不拆分
	if len(apes)+len(tollgates)+len(lanes) < batchSize {
		subscribeNotification.DeviceList = &model.APEListObjectEntity{APEObject: apes}
		subscribeNotification.TollgateObjectList = &model.TollgateListObjectEntity{TollgateObject: tollgates}
		subscribeNotification.LaneObjectList = &model.LaneListObjectEntity{LaneObject: lanes}
		subscribeNotificationList = append(subscribeNotificationList, subscribeNotification)
		subscribeNotificationModel := &model.SubscribeNotificationModel{
			SubscribeNotificationListObject: &model.SubscribeNotificationListObject{SubscribeNotificationObject: subscribeNotificationList},
		}
		i := []*model.SubscribeNotificationModel{subscribeNotificationModel}
		jsonData, _ := json.Marshal(i)
		logger.LOG_DEBUG(" 通知结构为 ：", string(jsonData))
		return i
	}
	if len(apes) > 0 {
		logger.LOG_WARN("共包含设备集合个数为 ：", len(apes))
		batchDeal(len(apes), batchSize, func(start, end int) {
			logger.LOG_WARN(fmt.Sprintf("拆分设备集合index为：start ：%d,end :%d", start, end))
			func(apes []*model.APE) {
				tempSubscribeNotification := *subscribeNotification
				subscribeNotification := &tempSubscribeNotification
				subscribeNotification.TriggerTime = times.Time2StrWithFormat(times.DateFormatNoSpan)
				subscribeNotification.DeviceList = &model.APEListObjectEntity{APEObject: apes}
				subscribeNotificationList = append(subscribeNotificationList, subscribeNotification)
			}(apes[start:end])
		})
	}
	if len(tollgates) > 0 {
		logger.LOG_WARN("共包含卡口集合个数为 ：", len(tollgates))
		batchDeal(len(tollgates), batchSize, func(start, end int) {
			logger.LOG_WARN(fmt.Sprintf("拆分卡口集合index为：start ：%d,end :%d", start, end))
			func(tollgates []*model.Tollgate) {
				tempSubscribeNotification := *subscribeNotification
				subscribeNotification := &tempSubscribeNotification
				subscribeNotification.TriggerTime = times.Time2StrWithFormat(times.DateFormatNoSpan)
				subscribeNotification.TollgateObjectList = &model.TollgateListObjectEntity{TollgateObject: tollgates}
				subscribeNotificationList = append(subscribeNotificationList, subscribeNotification)
			}(tollgates[start:end])
		})
	}
	if len(lanes) > 0 {
		logger.LOG_WARN("共包含车道集合个数为 ：", len(lanes))
		batchDeal(len(lanes), batchSize, func(start, end int) {
			logger.LOG_WARN(fmt.Sprintf("拆分车道集合index为：start ：%d,end :%d", start, end))
			func(lanes []*model.Lane) {
				tempSubscribeNotification := *subscribeNotification
				subscribeNotification := &tempSubscribeNotification
				subscribeNotification.TriggerTime = times.Time2StrWithFormat(times.DateFormatNoSpan)
				subscribeNotification.LaneObjectList = &model.LaneListObjectEntity{LaneObject: lanes}
				subscribeNotificationList = append(subscribeNotificationList, subscribeNotification)
			}(lanes[start:end])
		})
	}
	for _, subscribeNotification := range subscribeNotificationList {
		if nil == subscribeNotification.LaneObjectList && nil == subscribeNotification.DeviceList && nil == subscribeNotification.TollgateObjectList {
			continue
		}
		subscribeNotificationModel := &model.SubscribeNotificationModel{
			SubscribeNotificationListObject: &model.SubscribeNotificationListObject{SubscribeNotificationObject: []*model.SubscribeNotification{subscribeNotification}},
		}
		subscribeNotificationModels = append(subscribeNotificationModels, subscribeNotificationModel)
	}
	return subscribeNotificationModels
}

func FlushSubscribeStatusCache(s *model.SubscribeLib, subscribeStatus string, failDetail string) error {
	//修改数据库
	err := db.UpdateSubscribeStatusBySubscribeID(s.SubscribeID, subscribeStatus, failDetail)
	if err != nil {
		logger.LOG_ERROR("修改", s.Title, "订阅状态", subscribeStatus, "失败", err)
		return errors.New("修改订阅状态失败")
	}
	//修改内存
	if si, ok := cache.GetIns().SubscribeTasks.Load(s.SubscribeID); ok {
		s := si.(*model.SubscribeLib)
		status, _ := strconv.Atoi(subscribeStatus)
		s.SubscribeStatus = int32(status)
	}
	return nil
}
func FlushOperateTypeCache(s *model.SubscribeLib, operateType string, subscribeStatus string) error {
	//修改数据库
	err := db.UpdateOperateTypeBySubscribeID(s.SubscribeID, operateType, subscribeStatus)
	if err != nil {
		logger.LOG_ERROR("修改", s.Title, "subscribeStatus： ", subscribeStatus, "operateType： ", operateType, "失败", err)
		return errors.New("修改订阅状态失败")
	}
	//修改内存
	if si, ok := cache.GetIns().SubscribeTasks.Load(s.SubscribeID); ok {
		s := si.(*model.SubscribeLib)
		typeInt, _ := strconv.Atoi(operateType)
		s.OperateType = int32(typeInt)
		status, _ := strconv.Atoi(subscribeStatus)
		s.SubscribeStatus = int32(status)
	}
	return nil
}
func Deallane(laneIds []string) ([]*model.URMLane, error) {
	var urmLanes = make([]*model.URMLane, 0)
	if laneIds == nil || len(laneIds) <= 0 {
		logger.LOG_WARN("车道ids为空")
		return urmLanes, nil
	}
	//调用urm接口
	batchSize := 1000
	if context.GetInt("urm.batchSize") != 0 {
		batchSize = context.GetInt("urm.batchSize")
	}
	logger.LOG_WARN("请求urm,分批获取车道,size is ：", len(laneIds))
	var wait sync.WaitGroup
	var lock sync.Mutex
	var batchError error
	batchDeal(len(laneIds), batchSize, func(start, end int) {
		wait.Add(1)
		go func(laneIds []string) {
			var batches = make([]*model.URMLane, 0)
			_, _, _, e := util.Request(fmt.Sprintf(URM_URL_LANE, os.Getenv("URM_IP_PORT")), http.MethodPost, map[string]string{
				"Content-Type": "application/json",
			}, map[string]interface{}{
				"ids": laneIds,
			}, &batches)
			lock.Lock()
			urmLanes = append(urmLanes, batches...)
			lock.Unlock()
			if e != nil {
				logger.LOG_ERROR("请求urm失败", e)
				batchError = e
			}
			wait.Done()
		}(laneIds[start:end])
	})
	wait.Wait()
	if batchError != nil {
		return nil, batchError
	}
	logger.LOG_WARN("请求urm，共获取卡口 size is :", len(urmLanes))
	return urmLanes, nil
}
func Dealtollgate(tollgateIds []string) ([]*model.URMTollgate, error) {
	var urmTollgates = make([]*model.URMTollgate, 0)
	if tollgateIds == nil || len(tollgateIds) <= 0 {
		logger.LOG_WARN("卡口ids为空")
		return urmTollgates, nil
	}
	//调用urm接口
	batchSize := 1000
	if context.GetInt("urm.batchSize") != 0 {
		batchSize = context.GetInt("urm.batchSize")
	}
	logger.LOG_WARN("请求urm,分批获取卡口,size is ：", len(tollgateIds))
	var wait sync.WaitGroup
	var lock sync.Mutex
	var batchError error
	batchDeal(len(tollgateIds), batchSize, func(start, end int) {
		wait.Add(1)
		go func(tollgateIds []string) {
			var batchTollgates = make([]*model.URMTollgate, 0)
			_, _, _, e := util.Request(fmt.Sprintf(URM_URL_TOLLGATE, os.Getenv("URM_IP_PORT")), http.MethodPost, map[string]string{
				"Content-Type": "application/json",
			}, map[string]interface{}{
				"ids": tollgateIds,
			}, &batchTollgates)
			lock.Lock()
			urmTollgates = append(urmTollgates, batchTollgates...)
			lock.Unlock()
			if e != nil {
				logger.LOG_ERROR("请求urm失败", e)
				batchError = e
			}
			wait.Done()
		}(tollgateIds[start:end])
	})
	wait.Wait()
	if batchError != nil {
		return nil, batchError
	}
	logger.LOG_WARN("请求urm，共获取卡口 size is :", len(urmTollgates))
	return urmTollgates, nil
}
func DealEquipment(equipmentIds []string) ([]*model.URMEquipment, error) {
	var urmEquipments = make([]*model.URMEquipment, 0)
	if equipmentIds == nil || len(equipmentIds) <= 0 {
		logger.LOG_WARN("未绑定采集设备")
		return urmEquipments, nil
	}
	batchSize := 1000
	if context.GetInt("urm.batchSize") != 0 {
		batchSize = context.GetInt("urm.batchSize")
	}

	var lock sync.Mutex
	logger.LOG_WARN("请求urm,分批获取采集设备,size is ：", len(equipmentIds))
	var wait sync.WaitGroup
	var batchError error
	batchDeal(len(equipmentIds), batchSize, func(start, end int) {
		wait.Add(1)
		go func() {
			var batchEquipments = make([]*model.URMEquipment, 0)
			_, _, _, e := util.Request(fmt.Sprintf(URM_URL_EQUIPMENT, os.Getenv("URM_IP_PORT")), http.MethodPost, map[string]string{
				"Content-Type": "application/json",
			}, map[string]interface{}{
				"ids": equipmentIds[start:end],
			}, &batchEquipments)
			lock.Lock()
			urmEquipments = append(urmEquipments, batchEquipments...)
			lock.Unlock()
			if e != nil {
				logger.LOG_ERROR("请求urm失败", e)
				batchError = e
			}
			wait.Done()
		}()
	})
	wait.Wait()
	if batchError != nil {
		return nil, batchError
	}
	logger.LOG_WARN("请求urm，共获取采集设备 size is :", len(urmEquipments))
	return urmEquipments, nil
}

// 分页查询
func DealEquipmentPageList(equipmentIds []string, recordStartNo, pageRecordNum int) ([]*model.URMEquipment, error) {
	var urmEquipments = make([]*model.URMEquipment, 0)
	if len(equipmentIds) <= 0 {
		return urmEquipments, nil
	}
	//调用urm接口

	urmDevices := &model.URMDevice{}
	_, _, _, e := util.Request(fmt.Sprintf(URM_PAGE_RESOURCE_BY_PARAM, os.Getenv("URM_IP_PORT")), http.MethodPost, map[string]string{
		"Content-Type": "application/json",
	}, map[string]interface{}{
		"currentPage": recordStartNo/pageRecordNum + 1,
		"pageSize":    pageRecordNum,
		"type":        []int{2, 7, 11, 14}, //全量查询
		"ids":         equipmentIds,
	}, urmDevices)
	if e != nil {
		logger.LOG_ERROR("请求urm失败", e)
		return nil, e
	}
	urmEquipments = append(urmEquipments, urmDevices.Records...)
	logger.LOG_WARN("请求urm，共获取采集设备 size is :", len(urmEquipments))
	return urmEquipments, nil
}

// 处理数据的通知
func NotificationData(s *model.SubscribeLib) {
	//获取视图库信息
	if _, ok := cache.GetIns().SubscribeManagers.Load(s.ManagerID); !ok {
		logger.LOG_ERROR("通知任务所属视图库id未备案")
		return
	}
	subscribeManagerInterFace, _ := cache.GetIns().SubscribeManagers.Load(s.ManagerID)
	subscribeManager := subscribeManagerInterFace.(*model.SubscribeManagerEx)
	//获取任务列表
	taskIds := subscribeManager.TaskIds
	if taskIds == nil || len(taskIds) <= 0 {
		logger.LOG_ERROR("此视图库id未含有绑定的任务集合,视图库为:", subscribeManager.SubscribeManager.ManagerName)
		return
	}
	//taskIds := []string{"50ee68b740d2466582c2e0f17289b4e9"}
	//subscribeManager := &model.SubscribeManagerEx{}
	//任务详情集合
	var TasksInfos = make(map[string]map[string]interface{}, 0)
	var tasksError error
	//获取任务详细信息
	for _, taskId := range taskIds {
		taskInfo, e := GetTransmitTaskById(taskId)
		if e != nil {
			logger.LOG_ERROR("get TaskInfo error", e)
			tasksError = e
		}
		if taskInfo != nil && len(taskInfo) > 0 {
			TasksInfos[taskId] = taskInfo
		}
	}
	if tasksError != nil {
		return
	}
	//本级视图库id
	managerId := subscribeManager.SubscribeManager.SelfManagerID
	//修改任务
	for _, tasksInfo := range TasksInfos {
		e := updateTransmitTaskById(tasksInfo, s, subscribeManager, managerId)
		if e != nil {
			tasksError = e
			logger.LOG_ERROR("updateTask error", e)
		}
	}
	//更新缓存
	if tasksError == nil {
		//取消订阅
		if s.OperateType == 1 {
			_ = FlushSubscribeStatusCache(s, "1", "")
		} else {
			//完成一次订阅动作
			_ = FlushSubscribeStatusCache(s, "99", "")
		}
	}
}
func updateTransmitTaskById(tasksInfo map[string]interface{}, s *model.SubscribeLib, subscribeManager *model.SubscribeManagerEx, managerId string) error {

	//解析原有config
	configMap := make(map[string]interface{}, 0)
	if config, ok := tasksInfo["config"]; ok {
		e := json.Unmarshal([]byte(config.(string)), &configMap)
		if e != nil {
			logger.LOG_ERROR("解析config失败", e)
			return errors.New("解析config失败")
		}
	}
	var assignEquipmentIds = make([]string, 0)
	var assignTollgateIds = make([]string, 0)
	//订阅资源路径 是否为视图库,若为视图库不下发设备,下发相关配置，组件不通过设备id过滤
	resourceURI := s.Subscribe.ResourceURI
	if resourceURI != subscribeManager.SelfManagerID {
		configMap["msgfilter_filterResource"] = 0
		//若订阅的为具体设备，先匹配设备类型，修改参数
		//匹配设备类型
		resourceIds := strings.Split(resourceURI, ",")
		var apeIdMap = make(map[string]string)
		var tollgateIdMap = make(map[string]string)

		equipmentIds := subscribeManager.EquipmentIds
		if equipmentIds != nil && len(equipmentIds) > 0 {
			for _, apeId := range equipmentIds {
				apeIdMap[apeId] = apeId
			}
		}
		tollgateIds := subscribeManager.TollgateIds
		if tollgateIds != nil {
			for _, tollgateId := range tollgateIds {
				tollgateIdMap[tollgateId] = tollgateId
			}
		}
		for _, resourceId := range resourceIds {
			if _, ok := apeIdMap[resourceId]; ok {
				assignEquipmentIds = append(assignEquipmentIds, resourceId)
				continue
			}
			if _, ok := tollgateIdMap[resourceId]; ok {
				assignTollgateIds = append(assignTollgateIds, resourceId)
				continue
			}
			logger.LOG_WARN("订阅数据，未包含此资源id", resourceId)
		}
	} else {
		configMap["msgfilter_filterResource"] = 0
		assignEquipmentIds = subscribeManager.EquipmentIds
		assignTollgateIds = subscribeManager.TollgateIds
	}

	var subscribAddrs = make(map[string]string, 0)
	var subscribIds = make(map[string]string, 0)
	var subscribTitles = make(map[string]string, 0)
	var dataType int
	//查询此视图库下的所有未取消的数据订阅任务
	cache.GetIns().SubscribeTasks.Range(func(key string, value interface{}) (shouldContinue bool) {
		if _, ok := value.(*model.SubscribeLib); !ok {
			logger.LOG_ERROR(fmt.Sprintf("类型转换错误，需要*model.SubscribeLib 类型，类型为:%T", value))
			return true
		}
		subscribeLib := value.(*model.SubscribeLib)
		if subscribeLib.ManagerID != s.ManagerID {
			return true
		}
		if subscribeLib.OperateType == 1 {
			return true
		}
		details := strings.Split(subscribeLib.SubscribeDetail, ",")
		for _, detail := range details {
			if _, ok := DataType[detail]; !ok {
				logger.LOG_WARN("此订阅detail不是数据订阅，跳过处理:  ", detail)
				continue
			}
			dagType := DataType[detail]
			dataType |= dagType
			logger.LOG_DEBUG("s.ReceiveAddr is  ======", subscribeLib.ReceiveAddr)
			subscribAddrs[detail] = subscribeLib.ReceiveAddr
			logger.LOG_DEBUG("s.SubscribeID is  ======", subscribeLib.SubscribeID)
			subscribIds[detail] = subscribeLib.SubscribeID
			logger.LOG_DEBUG("s.Title is  ======", subscribeLib.Title)
			subscribTitles[detail] = subscribeLib.Title
		}
		return true
	})
	subscribAddrsJson, _ := json.Marshal(subscribAddrs)
	subscribIdsJson, _ := json.Marshal(subscribIds)
	subscribTitlesJson, _ := json.Marshal(subscribTitles)

	logger.LOG_DEBUG("1400client_subscribAddr is  ====== > ", string(subscribAddrsJson))
	logger.LOG_DEBUG("1400client_subscribId is  ====== > ", string(subscribIdsJson))
	logger.LOG_DEBUG("1400client_subscribTitle is  ====== > ", string(subscribTitlesJson))

	configMap["1400client_subscribAddr"] = string(subscribAddrsJson)
	configMap["1400client_subscribId"] = string(subscribIdsJson)
	configMap["1400client_subscribTitle"] = string(subscribTitlesJson)

	configMap["1400client_userIdentify"] = managerId
	configMap["1400client_isSubscribe"] = "true"
	configMap["msgfilter_dataType"] = dataType
	logger.LOG_DEBUG("msgfilter_dataType is  ====== > ", dataType)
	configJson, _ := json.Marshal(configMap)
	tasksInfo["config"] = util.BytesString(configJson)
	tasksInfo["dataType"] = configMap["msgfilter_dataType"]
	var equipmentBuff bytes.Buffer
	for _, assignEquipmentId := range assignEquipmentIds {
		equipmentBuff.WriteString(assignEquipmentId)
		equipmentBuff.WriteString(",")
	}

	//连云港，暂去设备自动同步绑定
	//equipments := strings.TrimRight(equipmentBuff.String(), ",")
	//tasksInfo["equipmentIds"] = equipments
	//var tollgateBuff bytes.Buffer
	//for _, assignTollgateId := range assignTollgateIds {
	//	tollgateBuff.WriteString(assignTollgateId)
	//	tollgateBuff.WriteString(",")
	//}
	//tollgates := strings.TrimRight(tollgateBuff.String(), ",")
	//tasksInfo["tollgateIds"] = tollgates
	//tasksInfo["equipmentNum"] = len(assignEquipmentIds)
	//tasksInfo["tollgateNum"] = len(assignTollgateIds)

	tasksInfo["updateTime"] = times.Time2Str()
	//调用resource-manager服务接口更新任务详情
	//var res = &ResponseWrap{}
	resBytes, _, _, e := util.Request(fmt.Sprintf(RESOURCE_MANAGER_TASK_UPDATE, os.Getenv("RESOURCE_MANAGER"),
		8116), http.MethodPost, map[string]string{
		"Content-Type": "application/json",
	}, tasksInfo, nil)
	res := &util.ResponseWrap{}
	e = json.Unmarshal(resBytes, res)
	if e != nil || res.Code != 200 {
		logger.LOG_WARN("更新任务失败", e)
		return errors.New("更新任务失败")
	}
	logger.LOG_INFO("更新转发任务成功", util.BytesString(resBytes))
	return nil
}

func GetTransmitTaskById(taskId string) (map[string]interface{}, error) {
	res := make(map[string]interface{}, 0)
	_, _, _, e := util.Request(fmt.Sprintf(RESOURCE_MANAGER_TASK_INFO, os.Getenv("RESOURCE_MANAGER"), 8116, taskId), http.MethodPost, map[string]string{
		"Content-Type": "application/json",
	}, nil, &res)
	if e != nil {
		logger.LOG_ERROR("获取任务详情失败", e)
		return nil, e
	}
	return res, nil
}
func batchDeal(count int, batchSize int, cb func(start, end int)) {
	start := 0
	for start < count {
		end := start + batchSize
		if end > count {
			end = count
		}
		cb(start, end)
		start += batchSize
	}
}

var DataType = map[string]int{
	constant.PERSON_INFO:  PERSON,
	constant.FACE_INFO:    FACE,
	constant.VEHICLE_INFO: VEHICLE,
	constant.NOMOTOR_INFO: NONVEHICLE,
	constant.HIK_ILLEGAL:  VEHICLE, //海康扩展车辆违法数据
}

const (
	FACE       = 1
	PERSON     = 2
	VEHICLE    = 4
	NONVEHICLE = 8
)
